package com.chukun.flink.table.api.sql;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @author chukun
 * @version 1.0.0
 * @description csv文件操作
 * @createTime 2022年06月03日 23:52:00
 */
public class FileSystemTemplate {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String sql = "create table csv_test " +
                "(name string, age int, city string) " +
                "with(" +
                "'connector'='filesystem', " +
                "'path'='E:/note/test.csv'," +
                "'format'='csv'," +
                "'csv.field-delimiter'='|')";

        tableEnv.executeSql(sql);

        Table table = tableEnv.sqlQuery("select name, age,city from csv_test");

        DataStream<Row> rowDataStream = tableEnv.toDataStream(table);

        rowDataStream.print("csv-test");

        String sinkSql =  "create table csv_sink_test " +
                "(name string, age int, city string) " +
                "with(" +
                "'connector'='filesystem', " +
                "'path'='E:/note/test_sink.csv'," +
                "'format'='csv'," +
                "'csv.field-delimiter'='|')";

        tableEnv.executeSql(sinkSql);

        tableEnv.executeSql("insert into csv_sink_test select name, age,city from csv_test");

        Table sinkTable = tableEnv.sqlQuery("select name, age, city from csv_sink_test");

        DataStream<Row> sinkDataStream = tableEnv.toDataStream(sinkTable);

        sinkDataStream.print("sink-from-csv");

        env.execute("FileSystemTemplate");
    }
}
